//
//  redis.h
//
//  Created by harrse on 5/5/17.
//  Copyright 2017 harrse. All rights reserved.
//

#ifndef __diylane_redis_h
#define __diylane_redis_h

#include <vector>
#include <string>

#ifdef __APPLE__
#pragma clang diagnostic push
#pragma clang system_header
#endif
#include <hiredis/hiredis.h>
#ifdef __APPLE__
#pragma clang diagnostic pop
#endif

#include <boost/thread.hpp>
#include "worker_pool.hpp"

namespace diylane
{
namespace redis
{

struct Reply
{
	redisReply *re;
	std::string error;

	Reply()
	{
		re = 0;
	}

	~Reply()
	{
		if (re)
		{
			::freeReplyObject(re);
			re = 0;
		}
	}

	void operator=(void *_reply)
	{
		re = (redisReply *)_reply;
	}

	redisReply *unwrap()
	{
		return re;
	}
};

struct Connection : pool::Worker
{
	redisContext *c = NULL;

	Connection()
	{
	}

	Connection(redisContext *ctx)
	{
		c = ctx;
	}

	redisContext *unwrap()
	{
		return c;
	}

	bool isAvailable()
	{
		if (busy || c == NULL || c->err != REDIS_OK)
		{
			return false;
		}
		return true;
	}
};

struct Server
{
	std::string host;
	std::string password;
	int database;
	int port;
};

struct Pool : pool::WorkerPool
{
	std::vector<Server> servers;
	int max_keepalive = 100; // seconds

	Pool()
	{
		name = "redis";
	}

	~Pool()
	{
	}

	// singleton helper
	static Pool &singleton(void *uniquePtr = NULL)
	{
		static Pool *singletonPointer = NULL;
		if (uniquePtr)
			singletonPointer = (Pool *)uniquePtr;
		return *singletonPointer;
	}

	void addServer(const char *host, const char *password, int database = 0, int port = 6379)
	{
		servers.push_back(Server{host, password, database, port});
	}

	/**
			 * Implementation of createOne_()
			 * @param error
			 * @return
			 */
	Connection *createOne_(std::string &error)
	{
		redisContext *ctx = NULL;
		struct timeval timeout = {5, 0};

		for (size_t i = 0; i < servers.size(); i++)
		{
			Server s = servers[i];
			ctx = ::redisConnectWithTimeout(s.host.c_str(), s.port, timeout);
			if (ctx == NULL)
			{
				error = "redis::Pool::Cannot allocate redis context!";
				return NULL;
			}

			// Check connection
			if (ctx->err)
			{
				continue;
			}

			::redisEnableKeepAlive(ctx);

			redisReply *reply = NULL;

			// Select database
			reply = (redisReply *)::redisCommand(ctx, "SELECT %d", s.database);
			if (reply == NULL)
			{
				::redisFree(ctx);
				error = "redis::Pool::Select database, no reply!";
				return NULL;
			}
			if (reply->type == REDIS_REPLY_ERROR)
			{
				if (strstr(reply->str, "NOAUTH") != reply->str)
				{
					error = "redis::Pool::Select database, ";
					error += reply->str;
					::redisFree(ctx);
					return NULL;
				}

				// Authenticate the connection
				reply = (redisReply *)::redisCommand(ctx, "AUTH %s", s.password.c_str());
				if (reply == NULL)
				{
					::redisFree(ctx);
					error = "redis::Pool::Auth, no reply!";
					return NULL;
				}
				if (reply->type == REDIS_REPLY_ERROR)
				{
					error = "redis::Pool::Auth, ";
					error += reply->str;
					::redisFree(ctx);
					return NULL;
				}
			}

			return new Connection(ctx);
		}
		return NULL;
	}

	Connection *acquire(std::string &errString)
	{
		return (Connection *)acquireWorker_(errString);
	}

	bool redisCommand(Reply &reply, const char *format, ...)
	{
		std::string err;

		Connection *conn = acquire(err);
		if (conn == NULL)
		{
			reply.error = "redis::Pool::redisCommand(), getConnection() returned NULL; ";
			reply.error += err;
			return false;
		}

		va_list ap;
		va_start(ap, format);
		reply = ::redisvCommand(conn->unwrap(), format, ap);
		va_end(ap);

		if (reply.unwrap() == NULL)
		{
			if (conn->unwrap()->err == REDIS_ERR_IO)
			{
				// TODO: reconnect
				reply.error = "redis::Pool::redisCommand(), no reply due to REDIS_ERR_IO, errno = ";
				reply.error += strerror(errno);
				release(conn);
				return false;
			}
			reply.error = "redis::Pool::redisCommand(), ";
			reply.error += conn->c->errstr;
			release(conn);
			return false;
		}

		if (reply.unwrap()->type == REDIS_REPLY_ERROR)
		{
			reply.error = "redis::Pool::redisCommand(), ";
			reply.error += reply.unwrap()->str;
			release(conn);
			return false;
		}

		release(conn);
		return true;
	}
};
}
}

#endif
